import logging
import os
import shutil
import warnings
from types import TracebackType
from typing import Any, Callable, Dict, List, Optional, Text, Type

import rasa.core.utils
import rasa.utils.io
from rasa.cli import utils
from rasa.cli.utils import bcolors
from rasa.constants import (
    DEFAULT_LOG_LEVEL,
    DEFAULT_LOG_LEVEL_LIBRARIES,
    ENV_LOG_LEVEL,
    ENV_LOG_LEVEL_LIBRARIES,
    GLOBAL_USER_CONFIG_PATH,
)

logger = logging.getLogger(__name__)


class TempDirectoryPath(str):
    """表示临时目录的路径。
    当用作上下文管理器时，它会在退出时擦除目录的内容。
    """

    def __enter__(self) -> "TempDirectoryPath":
        return self

    def __exit__(
        self,
        _exc: Optional[Type[BaseException]],
        _value: Optional[Exception],
        _tb: Optional[TracebackType],
    ) -> bool:
        if os.path.exists(self):
            shutil.rmtree(self)


def arguments_of(func: Callable) -> List[Text]:
    """返回函数“func”的参数作为名称列表。"""
    import inspect

    return list(inspect.signature(func).parameters.keys())


def read_global_config() -> Dict[Text, Any]:
    """Read global Rasa configuration."""
    # noinspection PyBroadException
    try:
        return rasa.utils.io.read_config_file(GLOBAL_USER_CONFIG_PATH)
    except Exception:
        # if things go south we pretend there is no config
        return {}


def set_log_level(log_level: Optional[int] = None):
    """将Rasa和Tensorflow的日志级别设置为:
        （1）提供的日志级别或环境变量'LOG_LEVEL'中指定的日志级别。
        （2）如果没有，则将使用默认日志级别。
    """
    import logging

    if not log_level:
        log_level = os.environ.get(ENV_LOG_LEVEL, DEFAULT_LOG_LEVEL)
        log_level = logging.getLevelName(log_level)

    logging.getLogger("rasa").setLevel(log_level)

    # 更新日志级别
    update_tensorflow_log_level()   # Tensorflow日志设置
    update_asyncio_log_level()      # python异步框架asyncio日志设置
    update_apscheduler_log_level()  # 附表日志级别
    update_socketio_log_level()     # socketio的日志级别设置

    os.environ[ENV_LOG_LEVEL] = logging.getLevelName(log_level)


def update_apscheduler_log_level() -> None:
    """更新附表的日志级别"""
    log_level = os.environ.get(ENV_LOG_LEVEL_LIBRARIES, DEFAULT_LOG_LEVEL_LIBRARIES)

    apscheduler_loggers = [
        "apscheduler",
        "apscheduler.scheduler",
        "apscheduler.executors",
        "apscheduler.executors.default",
    ]

    for logger_name in apscheduler_loggers:
        logging.getLogger(logger_name).setLevel(log_level)
        logging.getLogger(logger_name).propagate = False


def update_socketio_log_level() -> None:
    """socketio的日志级别设置"""
    log_level = os.environ.get(ENV_LOG_LEVEL_LIBRARIES, DEFAULT_LOG_LEVEL_LIBRARIES)

    socketio_loggers = ["websockets.protocol", "engineio.server", "socketio.server"]

    for logger_name in socketio_loggers:
        logging.getLogger(logger_name).setLevel(log_level)
        logging.getLogger(logger_name).propagate = False


def update_tensorflow_log_level() -> None:
    """将Tensorflow的日志级别设置为在环境变量'LOG_LEVEL_LIBRARIES'中指定的日志级别,
        变更Tensorflow有关的log配置
    """

    # 禁用libvinfer，tensorRT，cuda，AVX2和FMA warnings（CPU支持）。
    # 由于首次导入会引发一些警告，因此需要在首次导入之前设置此变量。
    os.environ["TF_CPP_MIN_LOG_LEVEL"] = "2"

    import tensorflow as tf

    log_level = os.environ.get(ENV_LOG_LEVEL_LIBRARIES, DEFAULT_LOG_LEVEL_LIBRARIES)

    if log_level == "DEBUG":
        tf_log_level = tf.compat.v1.logging.DEBUG
    elif log_level == "INFO":
        tf_log_level = tf.compat.v1.logging.INFO
    elif log_level == "WARNING":
        tf_log_level = tf.compat.v1.logging.WARN
    else:
        tf_log_level = tf.compat.v1.logging.ERROR

    tf.compat.v1.logging.set_verbosity(tf_log_level)
    logging.getLogger("tensorflow").propagate = False


def update_sanic_log_level(log_file: Optional[Text] = None):
    """将sanic的日志级别设置为在环境变量'LOG_LEVEL_LIBRARIES'中指定的日志级别。"""

    from sanic.log import logger, error_logger, access_logger

    log_level = os.environ.get(ENV_LOG_LEVEL_LIBRARIES, DEFAULT_LOG_LEVEL_LIBRARIES)

    logger.setLevel(log_level)
    error_logger.setLevel(log_level)
    access_logger.setLevel(log_level)

    logger.propagate = False
    error_logger.propagate = False
    access_logger.propagate = False

    if log_file is not None:
        formatter = logging.Formatter("%(asctime)s [%(levelname)-5.5s]  %(message)s")
        file_handler = logging.FileHandler(log_file)
        file_handler.setFormatter(formatter)

        logger.addHandler(file_handler)
        error_logger.addHandler(file_handler)
        access_logger.addHandler(file_handler)


def update_asyncio_log_level() -> None:
    """将asyncio协程的日志级别设置为在环境变量'LOG_LEVEL_LIBRARIES'中指定的日志级别"""

    log_level = os.environ.get(ENV_LOG_LEVEL_LIBRARIES, DEFAULT_LOG_LEVEL_LIBRARIES)
    logging.getLogger("asyncio").setLevel(log_level)


def obtain_verbosity() -> int:
    """Returns a verbosity level according to the set log level."""
    log_level = os.environ.get(ENV_LOG_LEVEL, DEFAULT_LOG_LEVEL)

    verbosity = 0
    if log_level == "DEBUG":
        verbosity = 2
    if log_level == "INFO":
        verbosity = 1

    return verbosity


def is_logging_disabled() -> bool:
    """Returns true, if log level is set to WARNING or ERROR, false otherwise."""
    log_level = os.environ.get(ENV_LOG_LEVEL, DEFAULT_LOG_LEVEL)

    return log_level == "ERROR" or log_level == "WARNING"


def sort_list_of_dicts_by_first_key(dicts: List[Dict]) -> List[Dict]:
    """Sorts a list of dictionaries by their first key."""
    return sorted(dicts, key=lambda d: list(d.keys())[0])


# noinspection PyUnresolvedReferences

def class_from_module_path(
    module_path: Text, lookup_path: Optional[Text] = None
) -> Any:
    """给定类的模块名称和路径，尝试检索该类，加载的类可用于实例化新对象。"""

    import importlib

    # 加载模块，如果无法加载模块将引发ImportError
    if "." in module_path:
        module_name, _, class_name = module_path.rpartition(".")
        m = importlib.import_module(module_name)

        # 获取类，如果找不到类，将引发AttributeError
        return getattr(m, class_name)

    else:
        module = globals().get(module_path, locals().get(module_path))
        if module is not None:
            return module

        # 最后一招：尝试从查找路径导入类
        if lookup_path:
            m = importlib.import_module(lookup_path)
            return getattr(m, module_path)
        else:
            raise ImportError(f"Cannot retrieve class from path {module_path}.")


def minimal_kwargs(
    kwargs: Dict[Text, Any], func: Callable, excluded_keys: Optional[List] = None
) -> Dict[Text, Any]:
    """仅返回函数所需的kwarg。
        不包括exception列表中包含的键。

    Args:
        kwargs:         所有可用的kwargs
        func:           应当被调用的函数
        excluded_keys:  从结果中排除的keys
    Returns:
        func接收的kwarg子集
    """
    # 排除的key
    excluded_keys = excluded_keys or []

    # func的参数
    possible_arguments = arguments_of(func)
    # 仅返回kwargs的参数
    return {
        k: v
        for k, v in kwargs.items()
        if k in possible_arguments and k not in excluded_keys
    }


def write_global_config_value(name: Text, value: Any) -> None:
    """阅读全局Rasa配置。"""

    try:
        os.makedirs(os.path.dirname(GLOBAL_USER_CONFIG_PATH), exist_ok=True)

        c = read_global_config()
        c[name] = value
        rasa.core.utils.dump_obj_as_yaml_to_file(GLOBAL_USER_CONFIG_PATH, c)
    except Exception as e:
        logger.warning(f"Failed to write global config. Error: {e}. Skipping.")


def read_global_config_value(name: Text, unavailable_ok: bool = True) -> Any:
    """Read a value from the global Rasa configuration."""

    def not_found():
        if unavailable_ok:
            return None
        else:
            raise ValueError(f"Configuration '{name}' key not found.")

    if not os.path.exists(GLOBAL_USER_CONFIG_PATH):
        return not_found()

    c = read_global_config()

    if name in c:
        return c[name]
    else:
        return not_found()


def mark_as_experimental_feature(feature_name: Text) -> None:
    """Warns users that they are using an experimental feature."""

    logger.warning(
        f"The {feature_name} is currently experimental and might change or be "
        "removed in the future 🔬 Please share your feedback on it in the "
        "forum (https://forum.rasa.com) to help us make this feature "
        "ready for production."
    )


def lazy_property(function: Callable) -> Any:
    """允许避免一遍又一遍地重新计算属性。

    结果存储在本地变量中。 在第一次调用该属性时，该属性的计算将发生一次。
    所有后续调用将使用存储在private属性中的值。
    """

    attr_name = "_lazy_" + function.__name__

    @property
    def _lazyprop(self):
        if not hasattr(self, attr_name):
            setattr(self, attr_name, function(self))
        return getattr(self, attr_name)

    return _lazyprop


def raise_warning(
    message: Text,
    category: Optional[Type[Warning]] = None,
    docs: Optional[Text] = None,
    **kwargs: Any,
) -> None:
    """发出带有合理默认值和彩色警告信息的`warnings.warn`。"""

    original_formatter = warnings.formatwarning

    def should_show_source_line() -> bool:
        if "stacklevel" not in kwargs:
            if category == UserWarning or category is None:
                return False
            if category == FutureWarning:
                return False
        return True

    def formatwarning(
        message: Text,
        category: Optional[Type[Warning]],
        filename: Text,
        lineno: Optional[int],
        line: Optional[Text] = None,
    ):
        """以标准方式格式化警告的功能。"""

        if not should_show_source_line():
            if docs:
                line = f"More info at {docs}"
            else:
                line = ""

        formatted_message = original_formatter(
            message, category, filename, lineno, line
        )
        return utils.wrap_with_color(formatted_message, color=bcolors.WARNING)

    if "stacklevel" not in kwargs:
        # 尝试为最常见的警告类别设置有用的默认值
        if category == DeprecationWarning:
            kwargs["stacklevel"] = 3
        elif category == UserWarning:
            kwargs["stacklevel"] = 2
        elif category == FutureWarning:
            kwargs["stacklevel"] = 3

    warnings.formatwarning = formatwarning
    warnings.warn(message, category=category, **kwargs)
    warnings.formatwarning = original_formatter
